package com.hubu.thread.pattern;
import java.util.LinkedList;
import java.util.Queue;
public class ProducerConsumer {
    static class Message {
        private Long messageId;
        private String messageInfo;

        public Message(Long messageId, String messageInfo) {
            this.messageId = messageId;
            this.messageInfo = messageInfo;
        }

        @Override
        public String toString() {
            return "Message{" +
                    "messageId=" + messageId +
                    ", messageInfo='" + messageInfo + '\'' +
                    '}';
        }
    }


    static class MessageQueueManager{
        private int currentIndex=0;
        private int queueSize;
        private MessageQueue[] messageQueues;
        public MessageQueueManager(int queueSize){
            this.queueSize=queueSize;
            messageQueues=new MessageQueue[queueSize];
            for(int i=0;i<queueSize;i++){
                messageQueues[i]=new MessageQueue(20000);
            }
        }
        //多个线程投递消息到消息队列中
        public void offerMessage(Message message){
            //多个线程同时获取索引
            //更新index加锁
            MessageQueue messageQueue=null;
            synchronized (this){
                if(currentIndex==messageQueues.length){
                    currentIndex=0;
                }
                messageQueue=messageQueues[currentIndex++];
            }

            //并发执行
            messageQueue.offer(message);
        }
    }
    public static void main(String[] args) {
        MessageQueueManager manager=new MessageQueueManager(10);
        Thread t1=new Thread(()->{
            for(int i=1;i<=5000;i++){
                Message message=new Message((long)i,"message"+i);
                manager.offerMessage(message);
            }
        });
        Thread t2=new Thread(()->{
            for(int i=5001;i<=10000;i++){
                Message message=new Message((long)i,"message"+i);
                manager.offerMessage(message);
            }
        });
        Thread t3=new Thread(()->{
            for(int i=10001;i<=15000;i++){
                Message message=new Message((long)i,"message"+i);
                manager.offerMessage(message);
            }
        });
        Thread t4=new Thread(()->{
            for(int i=15001;i<=20000;i++){
                Message message=new Message((long)i,"message"+i);
                manager.offerMessage(message);
            }
        });

        t1.start();
        t2.start();
        t3.start();
        t4.start();
        try {
            t1.join();
            t2.join();
            t3.join();
            t4.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for(MessageQueue queue: manager.messageQueues){
            for(Message message:queue.taskQueue){
                System.out.println(message);
            }
        }
    }
    static class MessageQueue{
        private Queue<Message> taskQueue=null;
        int maxSize=0;
        MessageQueue(int maxSize) {
            this.maxSize = maxSize;
            taskQueue = new LinkedList<>();
        }
        public void offer(Message message){
            synchronized (taskQueue){
                while(taskQueue.size()==maxSize){
                    try {
                        taskQueue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                taskQueue.offer(message);
                taskQueue.notifyAll();
            }
        }
        public Message poll() {
            synchronized (taskQueue) {
                while (taskQueue.isEmpty()) {
                    try {
                        taskQueue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                Message message = taskQueue.poll();
                taskQueue.notifyAll();
                return message;
            }
        }
    }
}